package com.favccxx.message.domain.message.adapter.kafka;

import com.favccxx.message.domain.message.MessageProtocol;
import com.favccxx.message.domain.message.adapter.CamelAdapter;
import com.favccxx.message.domain.message.entity.endpoint.KafkaEndPoint;
import com.favccxx.message.domain.message.entity.endpoint.MessageEndPoint;
import lombok.extern.slf4j.Slf4j;
import org.apache.camel.CamelContext;
import org.apache.camel.component.kafka.KafkaComponent;
import org.apache.camel.component.kafka.KafkaConfiguration;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class KafkaAdapter implements CamelAdapter {

    @Override
    public String getUri(CamelContext camelContext, MessageEndPoint endPoint) {
        if (endPoint instanceof KafkaEndPoint) {
            KafkaEndPoint kafkaEndPoint = (KafkaEndPoint) endPoint;

            KafkaComponent kafkaComponent = buildKafkaComponent(kafkaEndPoint);
            camelContext.addComponent(kafkaEndPoint.getCode(), kafkaComponent);

            StringBuffer sbUri = new StringBuffer();
            sbUri.append(kafkaEndPoint.getCode())
                    .append(":")
                    .append(kafkaEndPoint.getTopic())
                    .append("?brokers=")
                    .append(kafkaEndPoint.getBrokers());

            String uri = sbUri.toString();
            log.info("kafka uri: -> " + uri);

            return uri;
        }
        return null;
    }

    private KafkaComponent buildKafkaComponent(KafkaEndPoint kafkaEndPoint) {
        KafkaComponent kafkaComponent = new KafkaComponent();
        kafkaComponent.setBrokers(kafkaEndPoint.getBrokers());

        KafkaConfiguration kafkaConfiguration = new KafkaConfiguration();
        kafkaConfiguration.setAutoCommitEnable(kafkaEndPoint.isAutoCommitEnable());
        kafkaConfiguration.setGroupId(kafkaEndPoint.getGroupId());
        kafkaConfiguration.setAutoCommitIntervalMs(kafkaEndPoint.getAutoCommitIntervalMs());
        kafkaConfiguration.setKeyDeserializer(kafkaEndPoint.getKeyDeserializer());
        kafkaConfiguration.setValueDeserializer(kafkaEndPoint.getValueDeserializer());

        kafkaComponent.setConfiguration(kafkaConfiguration);

        return kafkaComponent;
    }

    @Override
    public String getDriverType() {
        return MessageProtocol.KAFKA.getName();
    }
}
